package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	"github.com/louloulin/dataflare/pkg/workflow"
	"github.com/louloulin/dataflare/pkg/workflow/connector"
	"github.com/louloulin/dataflare/pkg/workflow/processor"
)

func main() {
	fmt.Println("=== DataFlare Comprehensive Verification ===")

	// Create verification directory
	err := os.MkdirAll("verification/data", 0755)
	if err != nil {
		log.Fatalf("Failed to create verification directory: %v", err)
	}

	// Run verifications
	verifyWorkflowParsing()
	verifyConnectors()
	verifyProcessors()
	verifyLocalRuntime()

	fmt.Println("\n=== Verification Complete ===")
	fmt.Println("All tests passed successfully!")
}

// Verify workflow parsing
func verifyWorkflowParsing() {
	fmt.Println("\n=== Verifying Workflow Parsing ===")

	// Create test workflow file
	workflowPath := "verification/data/test-workflow.yaml"
	workflowContent := `
version: "1.0"
name: "test-workflow"
description: "Test workflow for verification"

sources:
  memory-source:
    type: memory
    config:
      data:
        - {"id": 1, "value": 10}
        - {"id": 2, "value": 20}
        - {"id": 3, "value": 30}

processors:
  transform-processor:
    type: transform
    config:
      function: |
        function process(data) {
          const obj = JSON.parse(data);
          obj.value = obj.value * 2;
          obj.processed = true;
          return JSON.stringify(obj);
        }

sinks:
  console-sink:
    type: console
    config:
      format: json

edges:
  - from: "memory-source"
    to: "transform-processor"
  - from: "transform-processor"
    to: "console-sink"
`
	err := os.WriteFile(workflowPath, []byte(workflowContent), 0644)
	if err != nil {
		log.Fatalf("Failed to create test workflow file: %v", err)
	}

	// Parse workflow
	fmt.Println("Parsing workflow...")
	definition, err := workflow.ParseFromFile(workflowPath)
	if err != nil {
		log.Fatalf("Failed to parse workflow: %v", err)
	}

	fmt.Printf("Successfully parsed workflow: %s (%s)\n", definition.Name, definition.Description)
	fmt.Printf("Sources: %d, Processors: %d, Sinks: %d, Edges: %d\n",
		len(definition.Sources), len(definition.Processors), len(definition.Sinks), len(definition.Edges))
}

// Verify connectors
func verifyConnectors() {
	fmt.Println("\n=== Verifying Connectors ===")

	// Get connector factory
	factory := connector.DefaultRegistry.GetFactory()

	// Test memory connector
	fmt.Println("Testing memory connector...")
	memorySource, err := factory.CreateSourceConnector("memory")
	if err != nil {
		log.Fatalf("Failed to create memory source connector: %v", err)
	}
	fmt.Printf("Memory source connector created: %T\n", memorySource)

	// Test file connector
	fmt.Println("Testing file connector...")
	fileSource, err := factory.CreateSourceConnector("file")
	if err != nil {
		log.Fatalf("Failed to create file source connector: %v", err)
	}
	fmt.Printf("File source connector created: %T\n", fileSource)

	// Test CSV connector
	fmt.Println("Testing CSV connector...")
	csvSource, err := factory.CreateSourceConnector("csv")
	if err != nil {
		log.Fatalf("Failed to create CSV source connector: %v", err)
	}
	fmt.Printf("CSV source connector created: %T\n", csvSource)

	// Test console sink
	fmt.Println("Testing console sink...")
	consoleSink, err := factory.CreateSinkConnector("console")
	if err != nil {
		log.Fatalf("Failed to create console sink connector: %v", err)
	}
	fmt.Printf("Console sink connector created: %T\n", consoleSink)

	// Test MySQL connector (if available)
	fmt.Println("Testing MySQL connector...")
	mysqlSource, err := factory.CreateSourceConnector("mysql")
	if err != nil {
		fmt.Printf("MySQL source connector not available: %v\n", err)
	} else {
		fmt.Printf("MySQL source connector created: %T\n", mysqlSource)
	}
}

// Verify processors
func verifyProcessors() {
	fmt.Println("\n=== Verifying Processors ===")

	// Test transform processor
	fmt.Println("Testing transform processor...")

	// Create test data
	testData := `{"id": 1, "value": 10}`

	// Create transform processor
	transformProcessor, err := processor.NewProcessor("transform")
	if err != nil {
		log.Fatalf("Failed to create transform processor: %v", err)
	}

	// Initialize processor
	err = transformProcessor.Init(map[string]interface{}{
		"function": `
			function process(data) {
				const obj = JSON.parse(data);
				obj.value = obj.value * 2;
				obj.processed = true;
				return JSON.stringify(obj);
			}
		`,
	})
	if err != nil {
		log.Fatalf("Failed to initialize transform processor: %v", err)
	}

	// Process data
	result, err := transformProcessor.Process([]byte(testData), nil)
	if err != nil {
		log.Fatalf("Failed to process data: %v", err)
	}

	fmt.Printf("Transform processor test successful:\n")
	fmt.Printf("  Input: %s\n", testData)
	fmt.Printf("  Output: %s\n", string(result))
}

// Verify local runtime
func verifyLocalRuntime() {
	fmt.Println("\n=== Verifying Local Runtime ===")

	// Create engine
	fmt.Println("Creating workflow engine...")
	engine, err := workflow.NewEngine()
	if err != nil {
		log.Fatalf("Failed to create engine: %v", err)
	}
	defer engine.Shutdown()

	// Parse workflow
	workflowPath := "verification/data/test-workflow.yaml"
	definition, err := workflow.ParseFromFile(workflowPath)
	if err != nil {
		log.Fatalf("Failed to parse workflow: %v", err)
	}

	// Deploy workflow
	ctx := context.Background()
	fmt.Println("Deploying workflow...")
	workflowID, err := engine.DeployWorkflow(ctx, definition)
	if err != nil {
		log.Fatalf("Failed to deploy workflow: %v", err)
	}
	fmt.Printf("Workflow deployed with ID: %s\n", workflowID)

	// Start workflow
	fmt.Println("Starting workflow...")
	runID, err := engine.StartWorkflow(ctx, workflowID, nil)
	if err != nil {
		log.Fatalf("Failed to start workflow: %v", err)
	}
	fmt.Printf("Workflow started with run ID: %s\n", runID)

	// Wait for workflow to complete
	fmt.Println("Waiting for workflow to complete...")
	maxWaitTime := 30 * time.Second
	startTime := time.Now()

	for {
		if time.Since(startTime) > maxWaitTime {
			fmt.Println("Workflow execution timed out")
			break
		}

		status, err := engine.GetWorkflowStatus(ctx, runID)
		if err != nil {
			log.Printf("Failed to get workflow status: %v", err)
			time.Sleep(1 * time.Second)
			continue
		}

		fmt.Printf("Workflow status: %s (Progress: %d%%)\n", status.Status, status.Progress)

		if status.Status == "completed" || status.Status == "failed" {
			break
		}

		time.Sleep(1 * time.Second)
	}

	fmt.Println("Local runtime verification completed")
}
